package com.neusoft.javafeatures9.juc;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.*;
import java.util.stream.Collectors;

public class CompletetableFutrueDemo01 {
    public static void main(String[] args) throws Exception {
        //为啥要用CompletableFuture,不直接用Future???
        ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newCachedThreadPool();
        Callable<String> task1 = () ->"任务1";
        Callable<String> task2 = () ->"任务2";
        Callable<String> task3 = () ->"任务3";
        Callable<String> task4 = () ->"任务4";
        //执行
        List<Future<String>> resultFutures = new CopyOnWriteArrayList<>();
        resultFutures.add(threadPool.submit(task1));
        resultFutures.add(threadPool.submit(task2));
        resultFutures.addAll(threadPool.invokeAll(List.of(task3, task4)));
        //获取执行结果
        List<String> results = resultFutures.stream().map(stringFuture -> {
            String value = "";
            try {
                value = stringFuture.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
            return value;
        }).toList();
        results.forEach(System.out::println);
        //线程池用完，记得关闭，不然程序一直卡在这里
        threadPool.shutdown();

        //测试1： 玩炮齐鸣案例
        //testCompletableFuture();
        //测试2：盖房经典案例，方法大全
        //testCompletableFutureStatic();
    }

    protected static void testCompletableFutureStatic() throws Exception {
        //任务之间串行执行，有错误之间退出  thenRun（void->void）、thenAccept(A->void)、thenApply (A-->B),其中（A线程的结果-->B线程的返回值）
        //任务之间串行执行，不管是否有错误，都携带错误继续执行 handle
        //两个任务A、B并行执行，接收other  thenCombine
        //两个任务A、B串行执行，接收other  thenCompose
        //两个任务A、B串行执行，接收other  thenAcceptBothAsync =(大概理解)整体串行 = thenCombine(并行，程序一开始就会执行   ) + thenApply(消费串行)
        //两个任务并行执行，谁快用谁的  .applyToEither().acceptEither().runAfterEither()， 三个只是重载action不一样
        //所有异步完成 whenCompleteAsync
        //异步发生错误 exceptionally

        //建造房子 1、一拨人找沙子 thenCombine
        //建造房子 2、一拨人找水泥 thenCombine
        //建造房子 3、一拨人找钢筋 （前面三波人并行） thenCombine
        //建造房子 4、若有材料准备齐了, 进行打地基，盖一层，然后盖二层（这里串行）。thenApply thenApply
        //建造房子 5、房子两层已经封顶，现在要贴瓷砖，找两个人去采购，谁采购回来的快，用谁的瓷砖。 applyToEither
        //建造房子 6、交付使用了 whenCompleteAsync、exceptionally
        ExecutorService executor = Executors.newCachedThreadPool();
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            System.out.printf("[第一波人]%s,正在找沙子%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[第一波人]%s,沙子寻找完毕%n", Thread.currentThread().getName());
            return "一拨人找沙子完成 thenCombine";
        }, executor);
        CompletionStage<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
            //暂停几秒钟线程
            System.out.printf("[第二波人]%s,正在找水泥%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(4); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[第二波人]%s,水泥寻找完毕%n", Thread.currentThread().getName());
        }, executor);
        CompletionStage<String> completableFuture3 = CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            System.out.printf("[第三波人]%s,正在找钢筋%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[第三波人]%s,钢筋寻找完毕%n", Thread.currentThread().getName());
            return "[第三波人]三拨人找钢筋完成 thenCombine";
        }, executor);
        //=================================开始盖一层和二层==============================

        completableFuture.
                thenCombineAsync(completableFuture2, (s, unused) -> "合并：1和2;").
                thenCombineAsync(completableFuture3, (s, s2) -> s + "合并：1和2和3;").
                whenComplete((s, throwable) -> {
                    if (throwable != null) {
                        System.err.println("error:" + throwable.getMessage());
                    } else {
                        //建房材料准备完成， 然后在进行开工仪式，然后进行盖房，然后进行浇筑房顶
                        //这里join和get是等价的，都是：等待任务完成 或者 任务结果

                        //completableFuture.join();
                        //completableFuture.get();
                        System.err.println("计算结果：" + s + ",材料准备完成，现在进行盖房了");

                        getOpenWorkStage(executor).
                                thenComposeAsync(s1 -> getHouseFuture1(), executor).
                                thenComposeAsync(s2 -> getHouseFuture2(), executor).
                                thenComposeAsync(s3 -> getWaterStage(executor), executor).
                                thenApplyAsync((prepareGoods)->{
                                    //瓷砖供应，需要：.applyToEither().acceptEither().runAfterEither() 进行改写
                                    //暂停几秒钟线程
                                    System.out.println("....");
                                    System.out.println("....");
                                    System.out.printf("[所有人]%s,两层房子已经盖好了，房顶也有浇筑完成，现正在招募2批发商进瓷砖，谁先到货，用谁的瓷砖%n", Thread.currentThread().getName());
                                    try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
                                    System.err.printf("[所有人]%s,瓷砖供应商已经确定了，%n", Thread.currentThread().getName());
                                    return "瓷砖供应商已经确定了.";
                                }, executor).thenRunAsync(new Runnable() {
                                    @Override
                                    public void run() {
                                        System.out.println("消费01：run...");
                                    }
                                }).thenAccept(new Consumer<Void>() {
                                    @Override
                                    public void accept(Void unused) {
                                        System.out.println("消费02：accept");
                                    }
                                }).thenApplyAsync(new Function<Void, Object>() {
                                    @Override
                                    public Object apply(Void unused) {
                                        System.out.println("消费03：apply");
                                        return null;
                                    }
                                }).handleAsync(new BiFunction<Object, Throwable, String>() {
                                    @Override
                                    public String apply(Object o, Throwable throwable) {
                                        System.out.println("消费04：apply haddle ");
                                        System.out.println("下面开始：thenAcceptBoth....");
                                        return "..";
                                    }
                                }).thenAcceptBoth(CompletableFuture.supplyAsync(new Supplier<String>() {
                                    @Override
                                    public String get() {
                                        //这里的方法，会最先执行
                                        System.out.println("开始制作瓷砖");
                                        return "supplier 供应瓷砖了...";
                                    }
                                }), new BiConsumer<String, String>() {
                                    @Override
                                    public void accept(String s, String s2) {
                                        System.out.println("thenAcceptBoth.... accept" + s + s2);
                                    }
                                })
                                ;
                    }
                }).
                exceptionally(e->{
                    System.out.println("-----exception: "+e.getCause()+"\t"+e.getMessage());
                    return e.getMessage();
                })
                ;
        //主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:暂停3秒钟线程
        try { TimeUnit.SECONDS.sleep(25); } catch (InterruptedException e) { e.printStackTrace(); }
        executor.shutdown();
    }


    private static CompletableFuture<String> getOpenWorkStage(Executor executor) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("..");
            System.out.printf("[所有人]%s,材料准备完成呢个，进行开工仪式%n", Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.err.printf("[所有人]%s,材料准备完成，开工仪式完成%n", Thread.currentThread().getName());
            return "进行正式的开工仪式,后面需要盖两层楼。";
        }, executor);
    }

    private static CompletableFuture<String> getHouseFuture1() {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("...");
            System.out.println("...");
            //暂停几秒钟线程
            System.out.printf("[所有人]%s,正在盖一层%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[所有人]%s,一层盖好了%n", Thread.currentThread().getName());
            return "【所有人】第一层建造完毕";
        });
    }


    private static CompletableFuture<String> getHouseFuture2() {
        return CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            System.out.printf("[所有人]%s,正在盖二层%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[所有人]%s,二层盖好了%n", Thread.currentThread().getName());
            return "【所有人】第二层建造完毕";
        });
    }


    private static CompletableFuture<String> getWaterStage(Executor executor) {
        return CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            System.out.printf("[所有人]%s,两层楼框架已经好了，正在准备浇筑顶部%n", Thread.currentThread().getName());
            try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
            System.err.printf("[所有人]%s,两层顶部水泥土已经浇筑完毕，%n", Thread.currentThread().getName());
            return "两层顶部水泥已经浇筑完毕.";
        }, executor);
    }


    protected static void testCompletableFuture() {
        CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
            System.err.printf("【*** 异步线程 ***】%s, 突然接到了上将军的紧急联系电话。%n", Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } // 当此异步线程执行完成之后，才会让所有的子线程恢复执行
            System.err.printf("【*** 异步线程 ***】%s, 接到了新的作战任务，更改炮兵的作战目标。%n", Thread.currentThread().getName());
        }); // 异步线程不用考虑返回值了
        // 下面利用多线程的概念模拟一个炮兵打跑的形式，因为所有的操作都需要有统一的口令
        for (int x = 0; x < 20; x++) {
            new Thread(()->{
                System.out.printf("【START】%s，炮兵就位，等待开炮命令发出。%n", Thread.currentThread().getName());
                try { // get()方法未收到数据则持续阻塞，收到数据则解除阻塞
                    future.get(); // 异步的操作是没有返回结果的
                    System.err.printf("〖END〗%s，收到开火命令，万炮齐鸣。%n", Thread.currentThread().getName());
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }, "炮兵 - " + x).start();
        }
        System.out.println("【★主线程★】所有的炮兵线程进入到了就绪状态，等待后续命令的发送。");
    }
}
